package com.atbeijing.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSONObject;
import com.atbeijing.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.SimpleDateFormat;

/**
 * 独立访客明细
 * uv数据
 */
public class UniqueVisitApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.设置检查点 略
        //TODO 3.从Kafka中读取数据
        //3.1 声明消费者主题以及消费者组
        String topic = "dwd_page_log";
        String groupId = "unique_visit_app_group";
        String sinkTopic = "dwm_unique_visit";
        //3.2 获取kafkaSource
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //kafka流
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);
        //转json
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(r -> JSONObject.parseObject(r));

        //TODO 4.按照设备id进行分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(r -> r.getJSONObject("common").getString("mid"));

        //TODO 5.对当前设备的访问情况进行过滤  UV
        SingleOutputStreamOperator<JSONObject> filteredDS = keyedDS.filter(new RichFilterFunction<JSONObject>() {
            //上次访问日期
            private ValueState<String> lastVisitDateState;
            private SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                sdf = new SimpleDateFormat("yyyy-MM-dd");
                //创建状态描述器
                ValueStateDescriptor<String> lastVisitDateStateDescriptor = new ValueStateDescriptor<>("lastVisitDateState", String.class);
                //状态配置项
                StateTtlConfig ttlConfig = StateTtlConfig
                        //状态存活时间,作用:24小时内访问多次算一次
                        .newBuilder(Time.days(1L))
                        //状态修改后,失效时间是否会被修改 默认值
                        //.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        //状态过期后,如果获取状态是否返回  默认值
                        //.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                        .build();
                //为状态描述器添加配置项
                lastVisitDateStateDescriptor.enableTimeToLive(ttlConfig);
                //通过状态描述器创建状态
                lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDescriptor);
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {
                //获取上级页面id
                String lastPageId = value.getJSONObject("page").getString("last_page_id");
                //判断是否有上级访问页面  如果有上级访问页面，直接将该日志过滤掉
                if (lastPageId != null && lastPageId.length() > 0) {
                    return false;
                }
                //获取状态的值 --上次访问日期
                String lastVisitDate = lastVisitDateState.value();
                //获取页面日志当前访问日期
                Long ts = value.getLong("ts");
                String curDate = sdf.format(ts);
                //如果某个机器没有lastPageId也就是第一次访问app页面,当天访问过,上次访问日期与本次访问日期相同表示重复访问
                if (lastVisitDate != null && lastVisitDate.length() > 0 && lastVisitDate.equals(curDate)) {
                    return false;
                } else {
                    //当天第一次访问,记录日期,数据保留
                    lastVisitDateState.update(curDate);
                    return true;
                }
            }
        });

        //TODO 6.将UV写回到Kafka的dwm层
        filteredDS.print("独立访客>>>");

        filteredDS
                .map(r -> r.toJSONString())
                .addSink(MyKafkaUtil.getKafkaSink(sinkTopic));

        env.execute();
    }
}
